
    /*
        该模块为用户提供服务
    */
#include "muduo/net/TcpConnection.h"
#include "muduo/prot/codec.h"
#include "muduo/prot/dispatcher.h"
#include "../mqcommon/helpper.hpp"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/thread_pool.hpp"
#include "../mqcommon/mq_message.pb.h" //消息的结构化描述在该部分使用protobuf生成
#include "../mqcommon/mq_prot.pb.h"

#include "mq_consumer.hpp"
#include <condition_variable>
namespace xgmq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;

    using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;
    using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const muduo::net::TcpConnectionPtr &conn,
                const ProtobufCodecPtr &codec)
            : _cid(UUIDHelper::uuid()), _conn(conn), _codec(codec) {}
        ~Channel()
        {
            basicCancel();
        }
        std::string cid()
        {
            return _cid;
        }
        void openChannel()
        {
            std::string rid = UUIDHelper::uuid();
            openChannelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            // 发送请求
            _codec->send(_conn, req);
            // 等待响应
            waitResponse(rid);
        }
        void closeChannel()
        {
            std::string rid = UUIDHelper::uuid();
            closeChannelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            // 发送请求
            _codec->send(_conn, req);
            // 等待响应
            waitResponse(rid);
        }
        // 声明交换机和删除交换机
        bool declareExchange(const std::string &ename,
                             ExchangeType etype,
                             bool edurable,
                             bool edelete,
                             google::protobuf::Map<std::string, std::string> &eargs)
        {
            // 构建请求
            std::string rid = UUIDHelper::uuid(); // 生成唯一的请求id
            declareExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_durable(edurable);
            req.set_exchange_type(etype);
            req.set_auto_delet(edelete);
            req.mutable_args()->swap(eargs);
            // 发送请求
            _codec->send(_conn, req);
            // 等待响应
            basicCommonResponsePtr bp = waitResponse(rid);
            // 返回
            return bp->ok();
        }
        bool deleteExchange(const std::string &ename)
        {
            // 构建请求
            std::string rid = UUIDHelper::uuid(); // 生成唯一的请求id
            deleteExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            // 发送请求
            _codec->send(_conn, req);
            // 等待响应
            basicCommonResponsePtr bp = waitResponse(rid);
            // 返回
            return bp->ok();
        }

        // 声明队列和删除队列
        bool declareQueue(const std::string &qname,
                          bool qdurable,
                          bool qexclusive,
                          bool qauto_delete,
                          google::protobuf::Map<std::string, std::string> &qargs)
        {
            // 构建请求
            std::string rid = UUIDHelper::uuid(); // 生成唯一的请求id
            declareQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_durable(qdurable);
            req.set_auto_delet(qauto_delete);
            req.set_exclusive(qexclusive);
            req.mutable_args()->swap(qargs);

            // 发送请求
            _codec->send(_conn, req);
            // 等待响应
            basicCommonResponsePtr bp = waitResponse(rid);
            // 返回
            return bp->ok();
        }
        bool deleteQueue(const std::string &qname)
        {
            // 构建请求
            std::string rid = UUIDHelper::uuid(); // 生成唯一的请求id
            deleteQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            // 发送请求
            _codec->send(_conn, req);
            // 等待响应
            basicCommonResponsePtr bp = waitResponse(rid);
            // 返回
            return bp->ok();
        }

        // 队列的绑定与解绑
        bool queueBind(const std::string &ename, const std::string &qname, const std::string &key)
        {
            // 构建请求
            std::string rid = UUIDHelper::uuid(); // 生成唯一的请求id
            queueBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_exchange_name(ename);
            req.set_binding_key(key);

            // 发送请求
            _codec->send(_conn, req);
            // 等待响应
            basicCommonResponsePtr bp = waitResponse(rid);
            // 返回
            return bp->ok();
        }
        bool queueUnBind(const std::string &ename, const std::string &qname)
        {
            // 构建请求
            std::string rid = UUIDHelper::uuid(); // 生成唯一的请求id
            queueUnBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_exchange_name(ename);
            // 发送请求
            _codec->send(_conn, req);
            // 等待响应
            basicCommonResponsePtr bp = waitResponse(rid);
            // 返回
            return bp->ok();
        }

        // 发布消息
        bool basicPublish(const std::string &ename, basicProperties *bp, const std::string &body)
        {
            // 构建请求
            std::string rid = UUIDHelper::uuid(); // 生成唯一的请求id
            basicPublishRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_body(body);
            if (bp != nullptr)
            {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_deliver_mode(bp->deliver_mode());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }
            // 发送请求
            _codec->send(_conn, req);
            // 等待响应
            basicCommonResponsePtr bc = waitResponse(rid);
            // 返回
            return bc->ok();
        }

        // 确认消息
        bool basicAck(const std::string &msg_id)
        {
            DLOG("消息处理完毕，给服务器发送确认消息请求\n");
            // 构建请求
            std::string rid = UUIDHelper::uuid(); // 生成唯一的请求id
            basicAckRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname);
            req.set_msg_id(msg_id);
            // 发送请求
            _codec->send(_conn, req);
            // 等待响应
            basicCommonResponsePtr bp = waitResponse(rid);
            // 返回
            return bp->ok();
        }

        // 订阅一个队列的消息
        bool basicConsume(const std::string &consume_tag, bool auto_ack, const std::string &qname, const ConsumerCallback &cb)
        {
            if (_consumer.get() != nullptr)
            {
                DLOG("当前信道已经订阅其他队列消息了\n");
                return false;
            }
            // 构建请求
            std::string rid = UUIDHelper::uuid(); // 生成唯一的请求id
            basicConsumeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_consume_tag(consume_tag);
            req.set_auto_ack(auto_ack);
            // 发送请求
            _codec->send(_conn, req);
            // 等待响应
            basicCommonResponsePtr bp = waitResponse(rid);
            if (bp->ok() == false)
            {
                DLOG("添加订阅失败\n");
            }
            _consumer = std::make_shared<Consumer>(consume_tag, qname, auto_ack, cb);
            // 返回
            return true;
        }

        // 取消订阅
        bool basicCancel()
        {
            // 说明该信道不是一个消费者
            if (_consumer.get() == nullptr)
            {
                return false;
            }
            // 构建请求
            std::string rid = UUIDHelper::uuid(); // 生成唯一的请求id
            basicCancelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname);
            req.set_consum_tag(_consumer->tag);
            // 发送请求
            _codec->send(_conn, req);
            // 等待响应
            basicCommonResponsePtr bp = waitResponse(rid);
            _consumer.reset();
            // 返回
            return bp->ok();
        }

    public:
        // 收到基础响应后，将该相应添加到哈希表中
        void putBasicResponse(const basicCommonResponsePtr &resp)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _basic_resp.insert(std::make_pair(resp->rid(), resp));
            _cond.notify_all();
        }
        // 收到消息响应后，通过消费者的回调函数处理
        void consume(const basicConsumeResponsePtr &resp)
        {
            if (_consumer.get() == nullptr)
            {
                DLOG("消息处理时，未找到订阅者...\n");
                return;
            }
            if (_consumer->tag != resp->consum_tag())
            {
                DLOG("收到推送消息的消费者标识，与当前信道的消费者标识不一致...\n");
                return;
            }
            _consumer->callback(resp->consum_tag(), resp->mutable_properties(), resp->body());
        }

    private:
        // 等待响应
        basicCommonResponsePtr waitResponse(const std::string &rid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock, [&rid, this]()
                       { return _basic_resp.find(rid) != _basic_resp.end(); });
            // 将得到的响应对象带出去
            basicCommonResponsePtr bp = _basic_resp[rid];
            _basic_resp.erase(rid);
            return bp;
        }

    private:
        std::string _cid;                   // 信道id
        ProtobufCodecPtr _codec;            // 协议处理句柄
        muduo::net::TcpConnectionPtr _conn; // 连接管理
        Consumer::ptr _consumer;
        std::mutex _mutex;
        std::condition_variable _cond;
        std::unordered_map<std::string, basicCommonResponsePtr> _basic_resp; // 请求id和响应的映射
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager() {}
        Channel::ptr create(const muduo::net::TcpConnectionPtr &conn,
                            const ProtobufCodecPtr &codec)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto ch = std::make_shared<Channel>(conn, codec);
            _channels.insert(std::make_pair(ch->cid(), ch));
            return ch;
        }
        void remove(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(cid);
        }
        Channel::ptr get(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if (it == _channels.end())
            {
                DLOG("所获取的信道不能存在...\n");
                return Channel::ptr();
            }
            return it->second;
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
}
